package com.zheng.flink.study.dataset.transformation;

import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple3;

import java.util.ArrayList;
import java.util.List;

/**
 * @author zhengbo
 * @date 2019/12/13
 */
public class FirstNTransformation {

    public static void main(String[] args) throws Exception {
        //tuple字段一次为 姓名 年龄 薪资
        List<Tuple3<String, Integer, Integer>> personInfoList = new ArrayList<>();

        personInfoList.add(new Tuple3<>("zhangyi", 29, 6000));
        personInfoList.add(new Tuple3<>("zhangsan", 29, 6500));
        personInfoList.add(new Tuple3<>("lisi", 25, 6000));
        personInfoList.add(new Tuple3<>("wangsu", 27, 6000));
        personInfoList.add(new Tuple3<>("zxxy", 29, 20000));


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<Tuple3<String, Integer, Integer>> dataSource = env.fromCollection(personInfoList);

        //打印出前两个 zhangyi zhangsan
        dataSource.first(2).print();

        System.out.println("----------------------");
        //先按工资分组 在再打印出前两个
        dataSource.groupBy(2).first(2).print();

        System.out.println("----------------------");
        //先按工资分组 再按年龄生序排序 再打印出前两个
        dataSource.groupBy(2).sortGroup(1, Order.ASCENDING).first(2).print();
    }
}
